/*
 * Copyright (c) 2009 - 2025 Deutsches Elektronen-Synchroton,
 * Member of the Helmholtz Association, (DESY), HAMBURG, GERMANY
 *
 * This library is free software; you can redistribute it and/or modify
 * it under the terms of the GNU Library General Public License as
 * published by the Free Software Foundation; either version 2 of the
 * License, or (at your option) any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Library General Public License for more details.
 *
 * You should have received a copy of the GNU Library General Public
 * License along with this program (see the file COPYING.LIB for more
 * details); if not, write to the Free Software Foundation, Inc.,
 * 675 Mass Ave, Cambridge, MA 02139, USA.
 */
package org.dcache.nfs.v4;

import java.net.InetSocketAddress;
import java.security.Principal;
import java.time.Clock;
import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.HexFormat;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

import javax.annotation.concurrent.GuardedBy;

import org.dcache.nfs.ChimeraNFSException;
import org.dcache.nfs.status.BadSeqidException;
import org.dcache.nfs.status.BadSessionException;
import org.dcache.nfs.status.BadStateidException;
import org.dcache.nfs.status.CompleteAlreadyException;
import org.dcache.nfs.status.ExpiredException;
import org.dcache.nfs.status.NoGraceException;
import org.dcache.nfs.status.SeqMisorderedException;
import org.dcache.nfs.status.StaleClientidException;
import org.dcache.nfs.util.Opaque;
import org.dcache.nfs.v4.xdr.clientid4;
import org.dcache.nfs.v4.xdr.seqid4;
import org.dcache.nfs.v4.xdr.sessionid4;
import org.dcache.nfs.v4.xdr.state_owner4;
import org.dcache.nfs.v4.xdr.stateid4;
import org.dcache.nfs.v4.xdr.verifier4;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/*
 *  with great help of William A.(Andy) Adamson
 */

public class NFS4Client {

    private static final Logger _log = LoggerFactory.getLogger(NFS4Client.class);

    private final AtomicInteger _stateIdCounter = new AtomicInteger(0);

    /*
     * from NFSv4.1 spec:
     *
     *
     * A server's client record is a 5-tuple:
     *
     * 1. co_ownerid The client identifier string, from the eia_clientowner structure of the EXCHANGE_ID4args structure
     * 2. co_verifier: A client-specific value used to indicate reboots, from the eia_clientowner structure of the
     * EXCHANGE_ID4args structure 3. principal: The RPCSEC_GSS principal sent via the RPC headers 4. client ID: The
     * shorthand client identifier, generated by the server and returned via the eir_clientid field in the
     * EXCHANGE_ID4resok structure 5. confirmed: A private field on the server indicating whether or not a client record
     * has been confirmed. A client record is confirmed if there has been a successful CREATE_SESSION operation to
     * confirm it. Otherwise it is unconfirmed. An unconfirmed record is established by a EXCHANGE_ID call. Any
     * unconfirmed record that is not confirmed within a lease period may be removed.
     *
     */

    /**
     * Variable length string that uniquely defines the client.
     */
    private final byte[] _ownerId;

    /**
     * Client side generated verifier that is used to detect client reboots.
     */
    private final verifier4 _clientVerifier;

    /**
     * Server side generated verifier that is used to detect retry.
     */
    private verifier4 _serverVerifier;

    /**
     * The RPCSEC_GSS principal sent via the RPC headers.
     */
    private final Principal _principal;
    /**
     * Client id generated by the server.
     */
    private final clientid4 _clientId;

    /**
     * A flag to indicate whether or not a client record has been confirmed.
     */
    private boolean _isConfirmed = false;

    /**
     * The sequence number used to track session creations.
     */
    private int _sessionSequence = 1;

    private final Map<stateid4, NFS4State> _clientStates = new ConcurrentHashMap<>();

    /**
     * sessions associated with the client
     */
    private final Map<sessionid4, NFSv41Session> _sessions = new HashMap<>();

    /**
     * The point in time of the last lease renewal.
     */
    private volatile Instant _lastLeaseUpdate;

    /**
     * Open Owners associated with client.
     */
    private final Map<Opaque, StateOwner> _owners = new HashMap<>();
    /*
     *
     * Client identification is encapsulated in the following structure:
     *
     * struct nfs_client_id4 { verifier4 verifier; opaque id<NFS4_OPAQUE_LIMIT>; };
     *
     * The first field, verifier is a client incarnation verifier that is used to detect client reboots. Only if the
     * verifier is different from that which the server has previously recorded the client (as identified by the second
     * field of the structure, id) does the server start the process of canceling the client's leased state.
     *
     * The second field, id is a variable length string that uniquely defines the client.
     *
     */

    /**
     * Client's {@link InetSocketAddress} seen by server.
     */
    private final InetSocketAddress _clientAddress;
    /**
     * Server's {@link InetSocketAddress} seen by client;
     */
    private final InetSocketAddress _localAddress;
    private ClientCB _cl_cb = null; /* callback info */

    /**
     * Time duration of a valid lease.
     */
    private final Duration _leaseTime;

    /**
     * A flag to indicate that the client already have reclaimed associated states.
     */
    private boolean _reclaim_completed;

    /**
     * Indicates that server needs a callback channel with this client.
     */
    private final boolean _callbackNeeded;

    /**
     * Highest NFSv4 minor version supported by the client.
     */
    private final int _minorVersion;

    /**
     * State handler which managers this client.
     */
    private final NFSv4StateHandler _stateHandler;

    /**
     * Clock to use for all time related operations.
     */
    private final Clock _clock;

    /**
     * List of listeners to be notified when client is disposed.
     */
    private final List<DisposeListener<NFS4Client>> _disposeListeners = new ArrayList<>();

    public NFS4Client(NFSv4StateHandler stateHandler, clientid4 clientId, int minorVersion,
            InetSocketAddress clientAddress, InetSocketAddress localAddress,
            byte[] ownerID, verifier4 verifier, Principal principal, Duration leaseTime, boolean calbackNeeded) {

        _stateHandler = stateHandler;
        _clock = _stateHandler.getClock();
        _ownerId = Arrays.copyOf(ownerID, ownerID.length);
        _clientVerifier = verifier;
        _serverVerifier = verifier4.valueOf(System.currentTimeMillis());
        _principal = principal;
        _clientId = clientId;

        _clientAddress = clientAddress;
        _localAddress = localAddress;
        _lastLeaseUpdate = _clock.instant();
        _leaseTime = leaseTime;
        _callbackNeeded = calbackNeeded;
        _minorVersion = minorVersion;
        _reclaim_completed = _minorVersion == 0; // no reclaim for NFSv4.0 clients
        _log.debug("New client: {}", this);
    }

    public void setCB(ClientCB cb) {
        _cl_cb = cb;
    }

    public ClientCB getCB() {
        return _cl_cb;
    }

    /**
     * Returns the highest NFSv4 minor version number supported by the client.
     */
    public int getMinorVersion() {
        return _minorVersion;
    }

    /**
     * Get client's long-hand unique identifier.
     *
     * @return client's unique identifier.
     */
    public byte[] getOwnerId() {
        return _ownerId;
    }

    /**
     *
     * @return client generated verifier
     */
    public verifier4 serverGeneratedVerifier() {
        return _serverVerifier;
    }

    /**
     *
     * @return client id generated by server
     */
    public clientid4 getId() {
        return _clientId;
    }

    public boolean clientGeneratedVerifierEquals(verifier4 verifier) {
        return _clientVerifier.equals(verifier);
    }

    public boolean serverGeneratedVerifierEquals(verifier4 verifier) {
        return _serverVerifier.equals(verifier);
    }


    public synchronized boolean isConfirmed() {
        return _isConfirmed;
    }

    public synchronized void setConfirmed() {
        _isConfirmed = true;
    }

    public boolean isLeaseValid() {
        return _lastLeaseUpdate.plus(_leaseTime).isAfter(_clock.instant());
    }

    /**
     * Update client's lease time if it not expired.
     *
     * @throws ExpiredException if difference between current time and last lease more than max_lease_time
     */
    public void updateLeaseTime() throws ChimeraNFSException {

        Instant curentTime = _clock.instant();
        var delta = Duration.between(_lastLeaseUpdate, curentTime);
        if (delta.compareTo(_leaseTime) > 0) {
            throw new ExpiredException("lease time expired: (" + delta + "): " + HexFormat.of().formatHex(_ownerId) +
                    " (" + _clientId + ").");
        }
        _lastLeaseUpdate = curentTime;
    }

    /**
     * sets client lease time with current time
     */
    public void refreshLeaseTime() {
        _lastLeaseUpdate = _clock.instant();
    }

    /**
     * re-initialize client
     */
    public synchronized void reset() {
        refreshLeaseTime();
        _isConfirmed = false;
        _serverVerifier = verifier4.valueOf(System.currentTimeMillis());
    }

    /**
     * Get the client's {@link InetSocketAddress} seen by server.
     *
     * @return client's address
     */
    public InetSocketAddress getRemoteAddress() {
        return _clientAddress;
    }

    /**
     * Get server's {@link InetSocketAddress} seen by the client.
     *
     * @return server's address
     */
    public InetSocketAddress getLocalAddress() {
        return _localAddress;
    }

    public int currentSeqID() {
        return _sessionSequence;
    }

    private NFS4State createState(StateOwner stateOwner, byte type, NFS4State openState) throws ChimeraNFSException {

        NFS4State state = new NFS4State(openState, stateOwner, _stateHandler.createStateId(this, type, _stateIdCounter
                .incrementAndGet()));
        if (openState != null) {
            openState.addDisposeListener(s -> {
                // remove and dispose derived states.
                NFS4State nfsState = _clientStates.get(state.stateid());
                if (nfsState != null) {
                    _log.debug("removing derived state {}", nfsState);
                    nfsState.tryDispose();
                }
                _clientStates.remove(state.stateid());
            });
        }
        _clientStates.put(state.stateid(), state);
        return state;
    }

    /**
     * Create a new open state.
     *
     * @param stateOwner state owner
     * @return new open state.
     * @throws ChimeraNFSException
     */
    public NFS4State createOpenState(StateOwner stateOwner) throws ChimeraNFSException {
        return createState(stateOwner, Stateids.OPEN_STATE_ID, null);
    }

    /**
     * Create a new lock state.
     *
     * @param stateOwner state owner
     * @param openState open state to derive from
     * @return new lock state.
     * @throws ChimeraNFSException
     */
    public NFS4State createLockState(StateOwner stateOwner, NFS4State openState) throws ChimeraNFSException {
        return createState(stateOwner, Stateids.LOCK_STATE_ID, openState);
    }

    /**
     * Create a new layout state.
     *
     * @param stateOwner state owner
     * @return new layout state.
     * @throws ChimeraNFSException
     */
    public NFS4State createLayoutState(StateOwner stateOwner) throws ChimeraNFSException {
        return createState(stateOwner, Stateids.LAYOUT_STATE_ID, null);
    }

    /**
     * Create a new delegation state.
     *
     * @param stateOwner state owner.
     * @return new delegation state.
     * @throws ChimeraNFSException
     */
    public NFS4State createDelegationState(StateOwner stateOwner) throws ChimeraNFSException {
        return createState(stateOwner, Stateids.DELEGATION_STATE_ID, null);
    }

    /**
     * Create a new directory delegation state.
     *
     * @param stateOwner state owner.
     * @return new directory delegation state.
     * @throws ChimeraNFSException
     */
    public NFS4State createDirDelegationState(StateOwner stateOwner) throws ChimeraNFSException {
        return createState(stateOwner, Stateids.DIR_DELEGATION_STATE_ID, null);
    }

    public NFS4State createServerSideCopyState(StateOwner stateOwner, NFS4State openState) throws ChimeraNFSException {
        return createState(stateOwner, Stateids.SSC_STATE_ID, openState);
    }

    public void releaseState(stateid4 stateid) throws ChimeraNFSException {

        NFS4State state = _clientStates.get(stateid);
        if (state == null) {
            throw new BadStateidException("State not known to the client: " + stateid);
        }
        state.disposeIgnoreFailures();
        _clientStates.remove(stateid);
    }

    public void tryReleaseState(stateid4 stateid) throws ChimeraNFSException {

        NFS4State state = _clientStates.get(stateid);
        if (state == null) {
            throw new BadStateidException("State not known to the client: " + stateid);
        }
        state.tryDispose();
        _clientStates.remove(stateid);
    }

    public NFS4State state(stateid4 stateid) throws ChimeraNFSException {
        NFS4State state = _clientStates.get(stateid);
        if (state == null) {
            throw new BadStateidException("State not known to the client: " + stateid);
        }
        return state;
    }

    @Override
    public String toString() {
        StringBuilder sb = new StringBuilder();
        sb.append(_clientAddress).append(":")
                .append(HexFormat.of().formatHex(_ownerId))
                .append("@")
                .append(_clientId)
                .append(":v4.").append(getMinorVersion());
        return sb.toString();
    }

    /**
     *
     * @return list of sessions created by client.
     */
    public Collection<NFSv41Session> sessions() {
        return _sessions.values();
    }

    public synchronized NFSv41Session createSession(int sequence, int cacheSize, int cbCacheSize, int maxOps,
            int maxCbOps) throws ChimeraNFSException {

        /*
         * For unconfirmed cleints server expects sequence number to be equal to value of eir_sequenceid that was
         * returned in results of the EXCHANGE_ID.
         */
        _log.debug("session for sequience: {}", sequence);
        if (sequence > _sessionSequence && _isConfirmed) {
            throw new SeqMisorderedException("bad sequence id: " + _sessionSequence + " / " + sequence);
        }

        if (sequence == _sessionSequence - 1 && !_isConfirmed) {
            throw new SeqMisorderedException("bad sequence id: " + _sessionSequence + " / " + sequence);
        }

        if (sequence == _sessionSequence - 1) {
            _log.debug("Retransmit on create session detected");
            sessionid4 sessionid = _stateHandler.createSessionId(this, sequence);
            return _sessions.get(sessionid);
        }

        if (sequence != _sessionSequence) {
            throw new SeqMisorderedException("bad sequence id: " + _sessionSequence + " / " + sequence);
        }

        sessionid4 sessionid = _stateHandler.createSessionId(this, _sessionSequence);
        NFSv41Session session = new NFSv41Session(this, sessionid, cacheSize, cbCacheSize, maxOps, maxCbOps);

        _sessions.put(sessionid, session);
        _sessionSequence++;

        if (!_isConfirmed) {
            _isConfirmed = true;
            _log.debug("set client confirmed");
        }

        return session;
    }

    public synchronized void removeSession(sessionid4 id) throws BadSessionException {
        NFSv41Session session = _sessions.remove(id);
        if (session == null) {
            throw new BadSessionException("session not found");
        }
    }

    public synchronized NFSv41Session getSession(sessionid4 id) throws BadSessionException {
        NFSv41Session session = _sessions.get(id);
        if (session == null) {
            throw new BadSessionException("session not found");
        }
        return session;
    }

    /**
     * Tell if there are any sessions owned by the client.
     *
     * @return true if client has at least one session.
     */
    public boolean hasSessions() {
        return !_sessions.isEmpty();
    }

    public Principal principal() {
        return _principal;
    }

    public boolean hasState() {
        return !_clientStates.isEmpty();
    }

    /**
     * Attach the state to the client.
     *
     * @param state to attach
     */
    public void attachState(NFS4State state) {
        _clientStates.put(state.stateid(), state);
    }

    /**
     * Detach a state from the client.
     *
     * @param state to detach
     */
    public void detachState(NFS4State state) {
        _clientStates.remove(state.stateid());
    }

    @GuardedBy("this")
    private void drainStates() {
        Iterator<NFS4State> i = _clientStates.values().iterator();
        while (i.hasNext()) {
            NFS4State state = i.next();
            state.disposeIgnoreFailures();
            i.remove();
        }
    }

    /**
     * Release resources used by this client if not released yet. Any subsequent call will have no effect.
     */
    public synchronized final void tryDispose() throws ChimeraNFSException {
        drainStates();
        Iterator<DisposeListener<NFS4Client>> i = _disposeListeners.iterator();
        while (i.hasNext()) {
            DisposeListener<NFS4Client> listener = i.next();
            listener.notifyDisposed(this);
            i.remove();
        }
    }

    /**
     * Release resources used by this client if not released yet. Ignore any errors.
     */
    public synchronized final void disposeIgnoreFailures() {
        drainStates();
        _disposeListeners.forEach(l -> {
            try {
                l.notifyDisposed(NFS4Client.this);
            } catch (ChimeraNFSException e) {
                _log.warn("failed to notify client dispose listener {} : {}", l, e.getMessage());
            }
        });
        _disposeListeners.clear();
    }

    /**
     * Indicates that client have reclaimed all states held before server reboot.
     *
     * @throws ChimeraNFSException
     */
    public synchronized void reclaimComplete() throws ChimeraNFSException {
        if (_reclaim_completed) {
            throw new CompleteAlreadyException("Duplicating reclaim");
        }
        _stateHandler.reclaimComplete(getOwnerId());
        _reclaim_completed = true;
    }

    /**
     * Indicates that client wants to perfor state reclaim operation.
     *
     * @throws ChimeraNFSException
     */
    public synchronized void wantReclaim() throws ChimeraNFSException {
        if (_reclaim_completed) {
            throw new NoGraceException("Already complete");
        }
        _stateHandler.wantReclaim(getOwnerId());
    }

    public synchronized boolean needReclaim() {
        return !_reclaim_completed;
    }

    public boolean isCallbackNeede() {
        return _callbackNeeded;
    }

    /**
     * Get and validate {@link StateOwner}. If owner does not exist a new owner will be created initial {@code seq}.
     *
     * @param owner client unique state owner
     * @param seq open sequence to validate
     * @return state owner
     * @throws BadSeqidException if sequence out of order.
     */
    public synchronized StateOwner getOrCreateOwner(byte[] owner, seqid4 seq) throws BadSeqidException {
        StateOwner stateOwner;
        if (_minorVersion == 0) {
            Opaque k = Opaque.forBytes(owner);
            stateOwner = _owners.get(k);
            if (stateOwner == null) {
                state_owner4 so = new state_owner4();
                so.clientid = _clientId;
                so.owner = owner;
                stateOwner = new StateOwner(so, seq.value);
                _owners.put(k, stateOwner);
            } else {
                stateOwner.acceptAsNextSequence(seq);
            }
        } else {
            // for minor version client id derived from session
            state_owner4 so = new state_owner4();
            so.clientid = _clientId;
            so.owner = owner;
            stateOwner = new StateOwner(so, 0);
        }
        return stateOwner;
    }

    /**
     * Remove {@link StateOwner}.
     *
     * @param owner client unique state owner
     */
    public synchronized void releaseOwner(byte[] owner) throws StaleClientidException {
        Opaque k = Opaque.forBytes(owner);
        StateOwner stateOwner = _owners.remove(k);
        if (stateOwner == null) {
            throw new StaleClientidException();
        }
    }

    /**
     * Add listener to be notified when client is disposed.
     */
    synchronized public void addDisposeListener(DisposeListener<NFS4Client> disposeListener) {
        _disposeListeners.add(disposeListener);
    }
}
